package com.shujia.state

import java.util

import org.apache.flink.api.common.functions.{RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object Demo3ListState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //设置并行度
    env.setParallelism(2)

    //读取socker数据
    //nc -lk 8888

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    val kvDS: DataStream[(String, Double)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(4), split(2).toDouble)
    })

    val keyByDS: KeyedStream[(String, Double), String] = kvDS.keyBy(_._1)


    val avgAgeDS: DataStream[(String, Double)] = keyByDS.map(new AvgMapFunction)

    avgAgeDS.print()

    env.execute()

  }

}

class AvgMapFunction extends RichMapFunction[(String, Double), (String, Double)] {

  //使用list state 保存每一个班级所有的年龄
  var listState: ListState[Double] = _

  override def open(parameters: Configuration): Unit = {
    val context: RuntimeContext = getRuntimeContext

    val listStateDesc = new ListStateDescriptor[Double]("ages", classOf[Double])

    listState = context.getListState(listStateDesc)
  }

  override def map(value: (String, Double)): (String, Double) = {
    val clazz: String = value._1
    val age: Double = value._2

    //将年龄保存到状态中
    listState.add(age)


    //获取所有的年龄计算平均值
    val ages: util.Iterator[Double] = listState.get().iterator()

    var count = 0
    var sum = 0.0

    while (ages.hasNext) {
      val a: Double = ages.next()
      sum += a
      count += 1
    }

    val avgAge: Double = sum / count

    (clazz, avgAge)
  }
}
